package com.xxxx.steamTest;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.util.Arrays;
import java.util.List;

/**
 * @program: day0316
 * @description:
 * @author: CoreDao
 * @create: 2021-03-18 15:01
 **/

public class Operator_Transform {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("javas");

        JavaSparkContext jsc = new JavaSparkContext(conf);
        JavaStreamingContext jssc = new JavaStreamingContext(jsc, Durations.seconds(5));

        //黑名单广播
        Broadcast<List<String>> broadcast = jsc.broadcast(Arrays.asList("zs", "lisi"));

        JavaReceiverInputDStream<String> dstream1 = jssc.socketTextStream("D-node01", 9999);

        //name address
        JavaPairDStream<String, String> nameToLine = dstream1.mapToPair(new PairFunction<String, String, String>() {

            @Override
            public Tuple2<String, String> call(String line) throws Exception {
                String[] split = line.split(" ");
                Tuple2<String, String> tuple2 = new Tuple2<>(split[0], line);
                System.out.println(tuple2);
                return tuple2;
            }
        });

        JavaDStream<String> dstream3 = nameToLine.transform(new Function<JavaPairRDD<String, String>, JavaRDD<String>>() {
            @Override
            public JavaRDD<String> call(JavaPairRDD<String, String> orgRDD) throws Exception {
                //rdd之外的代码没次都会打印
                System.out.println("++++++++");

                return orgRDD.filter(new Function<Tuple2<String, String>, Boolean>() {
                    @Override
                    public Boolean call(Tuple2<String, String> v1) throws Exception {
                        List<String> value = broadcast.value();
                        System.out.println("---------------filter------------------");

                        return value.contains(v1._1);
                    }
                }).map(new Function<Tuple2<String, String>, String>() {
                    @Override
                    public String call(Tuple2<String, String> v1) throws Exception {
                        System.out.println("---------------map--------------");
                        return v1._2;
                    }
                });
            }
        });
        dstream3.print();



        jssc.start();
        try {
            jssc.awaitTermination ();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        jssc.stop();
    }
}
